package com.walala.logic.resource;

import com.alibaba.fastjson.JSON;
import com.walala.logic.common.Global;
import com.walala.logic.common.RequestType;
import com.walala.logic.resource.bean.RequestMsg;
import com.walala.logic.zk.ZkClient;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 定时上报和获取服务器资源
 */
public class Resource {

	private static Logger logger = LoggerFactory.getLogger(Resource.class);

	// 调度服务器资源
	public static ConcurrentSkipListSet<String> DISPATCHER_SERVER_LIST = new ConcurrentSkipListSet<String>();

	// 逻辑服务器资源
	public static ConcurrentSkipListSet<String> LOGIC_SERVER_LIST = new ConcurrentSkipListSet<String>();

	static {
		// 初始化zk
		ZkClient.getInstance().initZkClient(Global.zk_addr);

		// 从zk获取调度服务器地址
		ZkClient.getInstance().getChildPath(Global.dispatcher_server_path, DISPATCHER_SERVER_LIST);
	}

	/**
	 * 定时拉取和上报资源
	 */
	public static void run() {

		// 接入层上报
		Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				report();
			}
		}, 0, 120000, TimeUnit.MILLISECONDS);

		// 逻辑服务器资源拉取
		Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				pulloutServer();
			}
		}, 0, 120000, TimeUnit.MILLISECONDS);

	}

	/**
	 * 上报逻辑层资源
	 */
	public static void report() {
		try {
			RequestMsg msg = new RequestMsg();
			msg.setIpPort(Global.server_ip + ":" + Global.server_port);
			msg.setRequestType(RequestType.REPORT_LOGIC);

			String jsonString = JSON.toJSONString(msg);

			if (DISPATCHER_SERVER_LIST.isEmpty()) {
				logger.error("无法获取调度服务器地址");
				return;
			} else {
				for (String address : DISPATCHER_SERVER_LIST) {
					String[] arr = address.split(":");
					String ip = arr[0];
					int port = Integer.parseInt(arr[1]);
					InetAddress address_ = InetAddress.getByName(ip);
					// 创建发送方的数据报信息
					DatagramPacket dataGramPacket = new DatagramPacket(jsonString.getBytes(),
							jsonString.getBytes().length, address_, port);

					DatagramSocket socket = new DatagramSocket(); // 创建套接字
					socket.setSoTimeout(5000);
					socket.send(dataGramPacket); // 通过套接字发送数据

					socket.close();
					logger.info("逻辑服务资源:[{}]已上报至调度服务:[{}]",Global.server_ip + ":" + Global.server_port,ip+":"+port);
				}
			}
		} catch (Exception e) {
			logger.error("逻辑层上报异常，error：", e);
		}
	}

	/**
	 * 拉取logic资源
	 */
	public static void pulloutServer() {
		RequestMsg msg = new RequestMsg();
		msg.setRequestType(RequestType.PULLOUT_LOGIC);
		String jsonString = JSON.toJSONString(msg);

		try {
			if (DISPATCHER_SERVER_LIST.isEmpty()) {
				logger.error("无法获取调度服务器地址");
				return;
			} else {
				for (String address : DISPATCHER_SERVER_LIST) {
					String[] arr = address.split(":");
					String ip = arr[0];
					int port = Integer.parseInt(arr[1]);
					InetAddress address_ = InetAddress.getByName(ip);
					// 创建发送方的数据报信息
					DatagramPacket dataGramPacket = new DatagramPacket(jsonString.getBytes(),
							jsonString.getBytes().length, address_, port);

					DatagramSocket socket = new DatagramSocket(); // 创建套接字
					socket.setSoTimeout(5000);
					socket.send(dataGramPacket); // 通过套接字发送数据

					// 接收服务器反馈数据
					byte[] buf = new byte[2048];
					DatagramPacket backPacket = new DatagramPacket(buf, buf.length);
					socket.receive(backPacket); // 接收返回数据
					String receiveMsg = new String(buf, 0, backPacket.getLength());
					logger.info("拉取到logic数据为:[{}]", receiveMsg);

					if (StringUtils.isNotBlank(receiveMsg)) {

						HashMap<String, String> map = (HashMap) JSON.parseObject(receiveMsg, Map.class);
						Set<String> keySet = map.keySet();
						for (String string : keySet) {
							LOGIC_SERVER_LIST.add(string);
						}
					}
					socket.close();
				}
			}
		} catch (Exception e) {
			logger.error("拉取logic异常，error：", e);
		}
	}

}
